/**
 * Pangolin IM Server
 * autor:   yiyefangzhou24
 * email:   yiyefangzhou24@qq.com
 * data:    21/10/1
 */

#include "pop_message.h"
#include "function.h"
#include "package.h"
#include "log.h"
#include "event2/event.h"
#include "hiredis.h"
#include "async.h"
#include "adapters/libevent.h"
#include "cJSON.h"
#include "base64.h"


/* 增加互斥锁 */
pthread_mutex_t SEND_MUTEX;
pthread_t mq_pids;

char config_chatserver_redishost[256] = { 0 };
int config_chatserver_redisport = 0;

/* 
*  [内部函数] 主要实现将消息解开后发送给相应的客户端
*/
int _send_to_client(int cmd , char * buff){
    pthread_mutex_lock(&SEND_MUTEX);
    //解析json格式数据
    cJSON * root = cJSON_Parse(buff);
    if(root == NULL){
        cJSON_Delete(root);
        LOG("_send_to_client : cJSON_Parse error");
        return -1;
    }
    //获得对方uid号
    cJSON * json_touid = cJSON_GetObjectItem(root , "to");
    if(json_touid == NULL){
        cJSON_Delete(root);
        LOG("_send_to_client : cJSON_GetObjectItem error");
        return -1;
    }

    //构建送的包结构
    int fd = redis_get_userfd(json_touid->valueint);
    char * str_package = malloc(sizeof(PACKAGE_HEAD) + strlen(buff) + 1);
    memset(str_package , 0 , sizeof(PACKAGE_HEAD) + strlen(buff) + 1);
    package_new(buff,strlen(buff),1,cmd, str_package);

    LOG("_send_to_client : senddata : fd:%d , package_body:%s" , fd , buff);

    //发送数据
    int ret = write(redis_get_userfd(json_touid->valueint),str_package,sizeof(PACKAGE_HEAD) + strlen(buff));

    free(str_package);
    cJSON_Delete(root);
    pthread_mutex_unlock(&SEND_MUTEX);
    return ret;
}

/* 
*  redis订阅消息回调函数，用于获取并格式化订阅的消息
*  当删除好友/发起（确认）添加好友/普通消息功能被调用，并发布了一条消息到消息队列中时，该函数被调用
*/
void _get_message(redisAsyncContext *c, void *r, void *priv) {
    redisReply *reply = r;  
    if (reply == NULL) return;  
    if ( reply->type == REDIS_REPLY_ARRAY && reply->elements == 3 ) {  
        if ( strcmp( reply->element[0]->str, "subscribe" ) != 0 ) { 
            //先进行base64解码
            LOG("_get_message ; enmsg : %s" , reply->element[2]->str);
            char msg_decode[PACKAGE_BUFF_MAX_LEN] = { 0 };
            base64_decode((void *)reply->element[2]->str , strlen(reply->element[2]->str) , (void *)msg_decode);
            LOG("_get_message ; msg : %s" , msg_decode);
            //LOG( "Received[%s] channel %s: %s\n",(char*)priv, reply->element[1]->str, reply->element[2]->str );
            //分离操作码和json数据
            char str_cmd[20] = { 0 };
            char str_data[PACKAGE_BUFF_MAX_LEN] = { 0 };

            int cur = (int )(strchr(msg_decode , ':') - msg_decode);
            if(cur <= 0) {
                LOG( "msg format error : %s",msg_decode );
                return;
            }
            strncpy(str_cmd , msg_decode , cur);
            int cmd = atoi(str_cmd);
            strcpy(str_data , msg_decode + cur + 1);
            if(_send_to_client(cmd , str_data) < 0){
                LOG("_send_to_client : error : %d " , errno);
            }
        }  
    }

}
void connectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        LOG("connectCallback : error : %s", c->errstr);
        return;
    }
}

void disconnectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        LOG("disconnectCallback : error : %s", c->errstr);
        return;
    }
}

/* 
*  线程函数
*  该消息队列是由redis的 发布/订阅 模式实现，在消息队列中监听订阅消息及消息队列pop功能
*  该模式为hiredis的异步模式
*/
void *create_pop_message_thread_handle(void *arg){
    signal(SIGPIPE, SIG_IGN);
    struct event_base *base = event_base_new();

    redisAsyncContext *c = redisAsyncConnect(config_chatserver_redishost,config_chatserver_redisport);
    if (c->err) {
        /* Let *c leak for now... */
        LOG("create_pop_message_thread_handle : error : %s", c->errstr);
        return 0;
    }

    redisLibeventAttach(c,base);
    redisAsyncSetConnectCallback(c,connectCallback);
    redisAsyncSetDisconnectCallback(c,disconnectCallback);
    redisAsyncCommand(c, _get_message, (char*) "sub" , "SUBSCRIBE msgqueue");
    event_base_dispatch(base);
    return 0;
}

/* 
*  消息队列线程函数。用于异步转发在线客户端之间的通信消息
*  可能使用消息队列的功能：删除好友/发起（确认）添加好友/普通消息
*/
void init_pop_message_thread(char * host , int port){
    strcpy(config_chatserver_redishost , host);
    config_chatserver_redisport = port;

    pthread_mutex_init(&SEND_MUTEX, NULL);
    //创建消息订阅线程
    pthread_create(&mq_pids, NULL, create_pop_message_thread_handle, NULL);

}